package ltd.dujiabao.dex.rabbitmq.exe;

import java.lang.reflect.InvocationTargetException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
import org.springframework.util.Assert;

import lombok.extern.slf4j.Slf4j;
import ltd.dujiabao.dex.core.DelayJobEntity;
import ltd.dujiabao.dex.core.registrar.DelayJobEntityHolder;
import ltd.dujiabao.dex.core.ExecutorType;
import ltd.dujiabao.dex.json.JsonUtils;
import ltd.dujiabao.dex.rabbitmq.domain.DelayMessage;
import ltd.dujiabao.dex.rabbitmq.domain.RabbitDefinition;

/**
 * @author : DuJiabao
 * @Project : common-delay-executor
 * @Time : 2024/3/28 21:46
 * @Desc :
 */
@Slf4j
public class RabbitDelayJobInitializer implements BeanPostProcessor, BeanFactoryAware {
    private ConfigurableBeanFactory beanFactory;
    private final RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry;
    private final RabbitListenerContainerFactory<?> rabbitListenerContainerFactory;
    private final Map<String, DelayJobEntity> handlers = new HashMap<>();

    public RabbitDelayJobInitializer(RabbitListenerEndpointRegistry rabbitListenerEndpointRegistry,
                                     RabbitListenerContainerFactory<?> rabbitListenerContainerFactory) {
        this.rabbitListenerEndpointRegistry = rabbitListenerEndpointRegistry;
        this.rabbitListenerContainerFactory = rabbitListenerContainerFactory;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (!(bean instanceof DelayJobEntityHolder)) {
            return bean;
        }

        Set<DelayJobEntity> delayJobEntities = ((DelayJobEntityHolder) bean).getDelayJobEntities();
        for (DelayJobEntity delayJobEntity : delayJobEntities) {
            if (delayJobEntity.getExecutorType() != ExecutorType.RABBITMQ) {
                continue;
            }
            Assert.notNull(beanFactory.getBean(delayJobEntity.getImplementation()), "Cannot found " + delayJobEntity.getImplementation());
            handlers.put(delayJobEntity.identity(), delayJobEntity);
            declareRabbitItems(delayJobEntity);
        }
        return bean;
    }

    private void declareRabbitItems(DelayJobEntity delayJobEntity) {
        String identity = delayJobEntity.identity();
        RabbitDefinition rabbitDefinition = RabbitDefinition.create(identity);

        // 注册死信交换
        registerDlxExchange(rabbitDefinition, beanFactory);
        // 注册死信队列
        registerDlxQueue(rabbitDefinition, beanFactory);
        // 注册死信绑定
        registerDlxBinding(rabbitDefinition, beanFactory);
        // 注册发射队列
        registerEmitQueue(rabbitDefinition, delayJobEntity, beanFactory);
        // 注册监听器
        registerRabbitListener(rabbitDefinition, delayJobEntity, beanFactory);
    }

    private void registerDlxExchange(RabbitDefinition rabbitDefinition, ConfigurableBeanFactory beanFactory) {
        String dlxExchangeName = rabbitDefinition.getDlxExchangeName();
        DirectExchange dlxExchange = new DirectExchange(dlxExchangeName, true, false);
        beanFactory.registerSingleton(dlxExchangeName, dlxExchange);
    }

    private void registerDlxQueue(RabbitDefinition rabbitDefinition, ConfigurableBeanFactory beanFactory) {
        String dlxQueueName = rabbitDefinition.getDlxQueueName();
        Queue dlxQueue = new Queue(dlxQueueName, true, false, false);
        beanFactory.registerSingleton(dlxQueueName, dlxQueue);
    }

    private void registerDlxBinding(RabbitDefinition rabbitDefinition, ConfigurableBeanFactory beanFactory) {
        String dlxBindingName = rabbitDefinition.getDlxBindingName();
        String dlxQueueName = rabbitDefinition.getDlxQueueName();
        String dlxExchangeName = rabbitDefinition.getDlxExchangeName();

        Binding binding = new Binding(dlxQueueName, Binding.DestinationType.QUEUE, dlxExchangeName, dlxBindingName, Collections.emptyMap());
        beanFactory.registerSingleton(dlxBindingName, binding);
    }

    private void registerEmitQueue(RabbitDefinition rabbitDefinition, DelayJobEntity delayJobEntity, ConfigurableBeanFactory beanFactory) {
        Map<String, Object> args = new HashMap<>(3);
        args.put("x-message-ttl", delayJobEntity.getTimeUnit().toMillis(delayJobEntity.getDelayTime()));
        args.put("x-dead-letter-exchange", rabbitDefinition.getDlxExchangeName());
        args.put("x-dead-letter-routing-key", rabbitDefinition.getDlxBindingName());

        String emitQueueName = rabbitDefinition.getEmitQueueName();
        Queue emitQueue = new Queue(emitQueueName, true, false, false, args);
        beanFactory.registerSingleton(emitQueueName, emitQueue);
    }

    private void registerRabbitListener(RabbitDefinition rabbitDefinition, DelayJobEntity delayJobEntity, ConfigurableBeanFactory beanFactory) {
        MethodRabbitListenerEndpoint endpoint = new MethodRabbitListenerEndpoint();
        try {
            endpoint.setMethod(this.getClass().getMethod("handle", Message.class));
            endpoint.setId(rabbitDefinition.getListenerName());
            endpoint.setBean(beanFactory.getBean(RabbitDelayJobInitializer.class));
            endpoint.setQueueNames(rabbitDefinition.getDlxQueueName());
            endpoint.setBeanFactory(beanFactory);
            endpoint.setMessageHandlerMethodFactory(createDefaultMessageHandlerMethodFactory(beanFactory));
            rabbitListenerEndpointRegistry.registerListenerContainer(endpoint, rabbitListenerContainerFactory, true);
        } catch (Exception exception) {
            log.error("error ", exception);
        }
    }

    static MessageHandlerMethodFactory createDefaultMessageHandlerMethodFactory(BeanFactory beanFactory) {
        DefaultMessageHandlerMethodFactory defaultFactory = new DefaultMessageHandlerMethodFactory();
        defaultFactory.setBeanFactory(beanFactory);
        DefaultConversionService conversionService = new DefaultConversionService();
        conversionService.addConverter(new BytesToStringConverter(StandardCharsets.UTF_8));
        defaultFactory.setConversionService(conversionService);
        defaultFactory.afterPropertiesSet();
        return defaultFactory;
    }

    private static class BytesToStringConverter implements Converter<byte[], String> {
        private final Charset charset;

        BytesToStringConverter(Charset charset) {
            this.charset = charset;
        }

        @Override
        public String convert(byte[] source) {
            return new String(source, this.charset);
        }
    }

    public void handle(Message message) throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
        String body = new String(message.getBody());
        DelayMessage delayMessage = JsonUtils.json2Bean(body, DelayMessage.class);
        Object bean = beanFactory.getBean(handlers.get(delayMessage.getIdentity()).getImplementation());
        delayMessage.getClazz().getMethod(delayMessage.getMethod(), delayMessage.getParameterTypes())
                .invoke(bean, delayMessage.getArguments());
    }


    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        this.beanFactory = (ConfigurableBeanFactory) beanFactory;
    }
}
